[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738
Draft
XiaoHongbo-Hope wants to merge 12 commits into
Draft
[python] Add range_join: shuffle-free Ray join for range-clustered tables#8738XiaoHongbo-Hope wants to merge 12 commits into
XiaoHongbo-Hope wants to merge 12 commits into
Conversation
XiaoHongbo-Hope
force-pushed
the
ray_range_join
branch
from
July 19, 2026 14:05
981697f to
daad7ae
Compare
…bles Cut the join-key space into ranges from per-file min/max stats in the manifest, then read and join each range in its own Ray task -- no global shuffle. Complements bucket_join for tables sorted/clustered by the join key rather than co-bucketed. Half-open ranges plus an in-memory range filter keep every matching pair produced exactly once regardless of range count; splits with no usable stats fall back to joining every range. Shared driver/worker helpers (table cache, snapshot pin, split read) are factored into join_common and reused by bucket_join.
…join output - Blocker: scan.plan() drops value stats, so every split looked unknown and the join always collapsed to one task. Read per-file min/max straight from the manifest (drop_stats=False), keyed by file name; stats never reach the workers. - Reject FLOAT/DOUBLE range keys (NaN falls out of every range while the hash join still matches NaN==NaN -> dropped matches) and TIMESTAMP_LTZ (naive manifest stats vs tz-aware column can't be compared). - Fix the collision check to catch a left key colliding with a right non-key column; document that the output keeps the left key names. - Cap the range count by an unknown-stats read budget so the every-range fallback re-reads at most one extra full scan.
The value_stats column names come from value_stats_cols when set, else the full schema -- the write_cols/schema_id detour resolved wrong names, so every split looked unknown and the join collapsed to one range. Read min/max via the same path files_table uses. Update the collision-message assertion.
value_stats.min/max_values are self-describing BinaryRows: they expose get_field(idx) and their own fields, not a .values list. The previous code read a non-existent .values attribute, so every split looked unknown and the join collapsed to one range. Index by the BinaryRow's own field names.
…fest pypaimon-written tables leave the manifest value stats empty (min/max live only in the parquet footer), so the manifest approach always saw unknown ranges and collapsed to one task. Read each split file's footer min/max via FileIO on the driver; the splits sent to workers still carry no stats. Verified locally on Python 3.11 (pyarrow+ray): ray_range_join_test 14 passed, ray_bucket_join_test 18 passed.
XiaoHongbo-Hope
force-pushed
the
ray_range_join
branch
from
July 22, 2026 09:18
cebd93b to
a8c7719
Compare
Under DATE->TIMESTAMP schema evolution, old files' footer bounds are dates and new files' are datetimes; comparing/sorting them raw raised TypeError. Coerce every footer min/max to the current join-key arrow type before use; a bound that can't be cast makes the split unknown (joins every range). Verified locally on Python 3.11: real DATE->TIMESTAMP evolution join now returns the correct result; ray_range_join_test 15 passed.
Coercing an old file's footer bounds to the current key type is unsafe when the cast doesn't preserve order: INT->STRING makes 5/100 into '5'/'100', but '42' > '100' as a string, so a coerced range silently drops the '42' row. Use a file's footer stats only when it stored the key in the current type; any other type (schema evolution) marks the split unknown (joins every range). Also drop the range predicate pushdown -- the reader can't compare a new-type bound against an evolved file's physical column; the in-memory clip does the exact, evolution-safe filtering. Verified locally on Python 3.11: INT->STRING and DATE->TIMESTAMP evolutions now return the full correct result; ray_range_join_test 16 passed.
XiaoHongbo-Hope
marked this pull request as ready for review
July 22, 2026 12:06
XiaoHongbo-Hope
marked this pull request as draft
July 22, 2026 12:20
- left_partitions/right_partitions ({column: value} on partition columns) prune
each side to those partitions before planning, so a join can target specific
partitions instead of whole tables.
- Read each split's parquet footer in a thread pool instead of serially, cutting
driver-side planning latency on many-file tables.
Verified locally on Python 3.11: partition-filtered and partitioned-table joins,
many-to-many correctness; ray_range_join_test 18 passed.
…nown ones The old budget only counted unknown-stats splits, but a known split whose [lo,hi] spans many ranges is also read once per range (no pushdown; read the whole split, clip in memory). On poorly clustered input that re-read could exceed a shuffle. Cap ranges by the actual total re-read -- rows times ranges each split overlaps -- halving num_ranges until it stays within a couple of full scans. Document that the feature only benefits clustered inputs. Verified locally: ray_range_join_test 18 passed (incl. the budget contract for wide/unknown splits), ray_bucket_join_test 18 passed.
…tes, validate inputs Review fixes: - Resolve the range key's footer column by field id, so a renamed/swapped (same-type) column reads its own stats instead of another column's. - Treat a PK table's non-PK range key as unknown: merge (aggregation/partial update) can move it past the file min/max, which would silently drop matches. - Bound re-read by file bytes, not row count, so a wide-row split can't be re-read hundreds of times within budget. - Validate partition keys (reject non-partition columns; None -> is_null) and num_ranges (int >= 1, capped at _MAX_RANGES even when passed explicitly). - Warn instead of silently degrading when a parquet footer read fails.
…e driver Fail fast for range keys that can't be range-partitioned instead of raising deep inside a Ray worker: nested/complex types (ARRAY/MAP/ROW/...) and both string forms of TIMESTAMP WITH LOCAL TIME ZONE (TIMESTAMP_LTZ(p) as well as ... WITH LOCAL TIME ZONE, only the latter was caught before).
…te all keys - P0: a range key under auth column-masking is rewritten at read time, so raw parquet footer min/max no longer bounds it -> treat as unknown (join every range, clip in memory), same fallback as a merge-rewritten non-PK key. - Validate every join key (not just the range key) up front and reject VARIANT as well as nested types, instead of failing as ArrowInvalid inside a worker. - Fix a stale test comment (footer, not manifest, stats).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Purpose
ray.data.joinshuffles both sides. When two tables are sorted/clustered by the join key, that shuffle is avoidable: cut the key space into ranges from per-file min/max stats in the manifest, then read and join each range in its own Ray task — no global shuffle. Complementsbucket_join(which needs co-bucketed tables) for the sorted-but-not-co-bucketed case.Half-open ranges plus an in-memory clip keep every matching pair produced exactly once regardless of range count; splits with no usable stats fall back to joining every range. Supports
on=orleft_on/right_on, inner join. Shared driver/worker helpers are factored intojoin_commonand reused bybucket_join.Tests
ray_range_join_test.py; flake8 clean. The exactly-once planning invariant was additionally verified over randomized range/skew/null trials.